#include "NetworkProtocol.h"
#include <arpa/inet.h>
#include <Exceptions.h>

using namespace CSNetworkLib;
using namespace CSApi;

void CleanThread(NetworkProtocol *protocol)
{
	FUNCTION_TRACK
	protocol->run();
	delete protocol;
	protocol = NULL;
}

NetworkProtocol::NetworkProtocol(socket_ptr s, uint32_t uniqueLimit) : m_socket(s) {
	FUNCTION_TRACK
	m_refBack = NULL;
	m_uniqueLimit = uniqueLimit;
	DBUG("Protocol Created Start - %p", this);
	m_stop = false;
    boost::thread m_thread(boost::bind(&CleanThread, this));
    m_seq = 0;
	AddHandler<NetworkProtocol>((uint16_t)PACKET_SYNC, &CSNetworkLib::NetworkProtocol::OnSync);
	DBUG("Protocol Created End - %p", this);
}

void NetworkProtocol::OnSync(const NetworkPacket &request)
{
	FUNCTION_TRACK
	uint8_t tmp = 0;
	if (request.Get_U8(0, tmp) && (1 == tmp))
	{
		Stop();
	}
}

NetworkProtocol::~NetworkProtocol() {
	FUNCTION_TRACK
	DBUG("Protocol Destroy Start - %p", this);
    m_stop = true;
	{
		NetworkPacket request(CSNetworkLib::PACKET_SYNC);
		request.Push_U8(1);
		SendPacket(request);
	}
	boost::system::error_code ec;
	{
		//boost::lock_guard<boost::mutex> lock(m_mutex);
		if (m_socket!=NULL) m_socket->shutdown(boost::asio::ip::tcp::socket::shutdown_both, ec);
		if (m_socket!=NULL) m_socket->close(ec);

		if (m_refBack!=NULL)
		{
			*m_refBack = NULL;
		}
		m_refBack = NULL;
		while(!m_handlers.empty())
		{
			delete m_handlers.begin()->second;
			m_handlers.erase(m_handlers.begin());
		}
	}
	DBUG("Protocol Destroy End - %p", this);
}

bool NetworkProtocol::RemoveHandler(uint16_t id) {
	FUNCTION_TRACK
    boost::lock_guard<boost::mutex> lock(m_mutex);
    if (m_handlers.find(id) != m_handlers.end())
        return false;
    delete m_handlers[id];
    m_handlers[id] = NULL;
    return m_handlers.erase(id) > 0;
}

bool NetworkProtocol::DispatchPacket(const NetworkPacket &packet) {
	FUNCTION_TRACK
	delegate_ptr tmp = NULL;
	{
		boost::lock_guard<boost::mutex> lock(m_mutex);
		std::map<uint16_t, delegate_ptr>::iterator it = m_handlers.find(packet.GetId());
		if (m_handlers.end() == it)
			return false;
		tmp = it->second;
	}
    tmp->Invoke(packet);
    return true;
}

bool NetworkProtocol::SendPacketUnique(const NetworkPacket &packet) {
	FUNCTION_TRACK

	if (m_unique.size()>m_uniqueLimit)
	{
		m_unique.clear();
	}
	uint32_t crc32 = packet.GetCRC32();
	{
		boost::lock_guard<boost::mutex> lock(m_mutex);
		if ( m_unique.find(crc32) != m_unique.end())
		{
			return true;
		}
	}

	if (SendPacket(packet))
	{
		boost::lock_guard<boost::mutex> lock(m_mutex);
		m_unique.insert(crc32);
		return true;
	}
	return false;
}

bool NetworkProtocol::SendPacket(const NetworkPacket &packet) {
	FUNCTION_TRACK
	if (NULL == m_socket) return false;
	DBUG("%p, SendPacket = Start", this);
    uint8_t tmp[BUFFOR_LENGTH];
    if (0 == packet.GetSeq())
	{
        boost::lock_guard<boost::mutex> lock(m_mutex);
        ++m_seq;
        if (m_seq > 0x0FFFFFFF)
            m_seq = 1;
        packet.SetSeq(m_seq);
		DBUG("%p, SendPacket = SetSeq(%d)", this, m_seq);
    }

    packet.SetWaitingForResponse(false);
    if (!packet.Serialize(tmp, BUFFOR_LENGTH)) {
		ERR("%p, SendPacket = Failed to serialize", this);
        return false;
    }

    boost::system::error_code error;
    {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        boost::asio::write((*m_socket), boost::asio::buffer(tmp, packet.GetSize()), error);
    }
    if (boost::system::errc::success != error)
	{
		Stop();
		ERR("%p, SendPacket = Failed to write - %s", this, error.message().c_str());
        return false;
	}
	DBUG("%p, SendPacket = Packet send", this);
    return true;
}

bool NetworkProtocol::SendPacketWait(const NetworkPacket &packet, NetworkPacket &response, uint32_t wait_time_in_ms) {
	FUNCTION_TRACK
    uint8_t tmp[BUFFOR_LENGTH];
    uint32_t seq = 0;
    {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        ++m_seq;
        if (m_seq > 0x0FFFFFFF)
            m_seq = 1;
        seq = m_seq;
        packet.SetSeq(m_seq);
		DBUG("Messsage with Seq ID set to %d", seq);
    }

    packet.SetResponse(false);
    packet.SetWaitingForResponse(true);
    if (!packet.Serialize(tmp, BUFFOR_LENGTH)) {
        return false;
    }
    TResponseWaitPacket ret;
    ret.processed = false;
    ret.response = &response;
    boost::system::error_code error;
    {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        m_waitList[seq] = &ret;
        boost::asio::write((*m_socket), boost::asio::buffer(tmp, packet.GetSize()), error);
    }
    if (boost::system::errc::success != error) {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        m_waitList.erase(seq);
		DBUG("Error durring sending message: %s", error.message().c_str());
		Stop();
        return false;
    }
    ret.Wait(wait_time_in_ms);
    if (!ret.processed) {
        boost::lock_guard<boost::mutex> lock(m_mutex);
        m_waitList.erase(seq);
		DBUG("Messsage with Seq %d not received", seq);
    } else
	{
		if ((uint32_t)PACKET_EXCEPTION == response.GetId())
		{
			uint8_t e = 0;
			response.Get_U8(0, e);
			std::string msg(response.GetString(1));
			ERR("Exception received: %d (%s)", e, msg.c_str());
			switch((EExceptionId)e)
			{
				case EException_WrongArgument : throw CSException_WrongArgument(msg);
				case EException_AuthenticationFailed : throw CSException_AuthenticationFailed(msg);
				case EException_InvalidPath : throw CSException_InvalidPath(msg);
				case EException_StorageOffline : throw CSException_StorageOffline(msg);
				case EException_FileNotExits : throw CSException_FileNotExits(msg);
				case EException_FolderNotExits : throw CSException_FolderNotExits(msg);
				case EException_FileExits : throw CSException_FileExits(msg);
				case EException_FolderExits : throw CSException_FolderExits(msg);
				case EException_InvalidType : throw CSException_InvalidType(msg);
				case EException_InvalidOperation: throw CSException_InvalidOperation(msg);
                case EException_OperationFailed: throw CSException_OperationFailed(msg);
				default : throw CSException_Unknown(msg);
			}
		}
		else
		{
			DBUG("Messsage with Seq %d received", seq);
		}

	}
    return ret.processed;
}

bool NetworkProtocol::SendExceptionResponse(EExceptionId exception, const std::string &message, const NetworkPacket& request)
{
	if (!request.IsWaitingForResponse())
		return false;
	NetworkPacket response(request, PACKET_EXCEPTION);
	response.Push_U8((uint8_t)exception);
	response.Push_String(message.c_str());
	return SendPacket(response);
}

void NetworkProtocol::run() {
	FUNCTION_TRACK
	{DBUG("Run thread started - %p", this);
    uint8_t tmp[BUFFOR_LENGTH];
    boost::system::error_code error;
    size_t length = 0;
    uint32_t size = 0, seq = 0;
    CSNetworkLib::NetworkPacket recive_packet;
	TResponseWaitPacket *pret = NULL;
    while (!m_stop) {

        if(m_socket->available() == 0)
        {
            boost::this_thread::sleep(boost::posix_time::milliseconds(100));
            continue;
        }

        recive_packet.Clear();
        length = m_socket->read_some(boost::asio::buffer(tmp, sizeof (uint32_t)), error);
		if (m_stop) break;
        if (error == boost::asio::error::eof) {
            ERR("Connection closed cleanly by peer.");
            break;
        } else if (error) {
            ERR("Error on reciving message (%s).", error.message().c_str());
            break;
        }

        if (length != sizeof (uint32_t)) {
            ERR("Wrong size, less than 4 bytes of header");
            break;
        }

        size = switchEndian<uint32_t>((*((uint32_t *) &tmp[0])), false);
        if ((size > BUFFOR_LENGTH) || (size < sizeof (uint32_t) + sizeof (uint32_t) + sizeof (uint16_t) + sizeof (uint16_t))) {
            ERR("Wrong packet - size (%d)", size);
            break;
        }
		DBUG("%p = Reading %d bytes from socket as one packet", this, size);
        size -= sizeof (uint32_t);
		if (m_stop) break;
        length = m_socket->read_some(boost::asio::buffer(&tmp[sizeof (uint32_t) ], size), error);
		if (m_stop) break;
        if (error == boost::asio::error::eof) {
            ERR("Connection closed cleanly by peer.");
            break;
        } else if (error) {
            ERR("Error on reciving message (%s).", error.message().c_str());
            break;
        }

        if (length != size) {
            ERR("Wrong packet - not read all");
            break;
        }

		seq = switchEndian<uint32_t>((*((uint32_t *) &tmp[8])), false);

        if ((seq & 0x10000000) == 0x10000000) {
			DBUG("Response received with Seq %d", seq & 0x0FFFFFFF);
            boost::lock_guard<boost::mutex> lock(m_mutex);
            std::map<uint32_t, TResponseWaitPacket* >::iterator it = m_waitList.find(seq & 0x0FFFFFFF);
            if (it != m_waitList.end()) {
				pret = (it->second);
				m_waitList.erase(it);

                if (pret->response->UnSerialize(tmp, size + sizeof (uint32_t)))
				{
					pret->NotifyAll();
				}
				else
				{
					ERR("Recived response message what can't be decrypted");
					break;
				}
                continue;
            }
        }

		if (!recive_packet.UnSerialize(tmp, size + sizeof (uint32_t))) {
            ERR("Recived message what can't be decrypted");
            break;
        }
		if (m_stop) break;
        if (!DispatchPacket(recive_packet)) {
            WARN("Packet with unknown ID received - %d", recive_packet.GetId());
        }
    }
	DBUG("Run thread finished - %p", this);
	m_stop = true;}
}